package com.we.risk.phoneRecord.var.dimuser;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.*;

import static com.we.risk.phoneBook.common.Variables.*;

public class DimCallRecordUserInfo {
    public static final String INPUT_KEY_BY = "uid";
    public static final String OUTPUT_KEY_BY = "user_key";
    public static final String SPLITWORD = "#";
    public static final String RELEASEPROP =
            "risk/phonerecord/var/dim_user_label_kfk_prod.properties";
    public static final String ADMUSERLABELPROP =
            "risk/phonerecord/var/user_info.properties";

    public static Logger LOG = LoggerFactory.getLogger(DimCallRecordUserInfo.class);

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();
            String srckfkOffset = paramReader.getTmpKfkOffset();

            Properties consumProp = new Properties();
            consumProp.setProperty("bootstrap.servers", srcKafkaBootStrapServer);
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty("auto.offset.reset", srckfkOffset);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaSource<String> dimUserlabelSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> dimUserLabelInput =
                    env.fromSource(
                                    dimUserlabelSource,
                                    WatermarkStrategy.noWatermarks(),
                                    "PRdimuserlabel")
                            .uid("AdmUserLabel_input");

            env.registerCachedFile(Stage2GlobalSetting.V1DATA, Stage2GlobalSetting.V1CATCHFILENAME);
            env.registerCachedFile(
                    Stage2GlobalSetting.V1CLEANDATA, Stage2GlobalSetting.V1CLEANCATCHFILENAME);

            SingleOutputStreamOperator<String> out =
                    dimUserLabelInput
                            .keyBy(
                                    new KeySelector<String, Long>() {
                                        @Override
                                        public Long getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getLong(INPUT_KEY_BY);
                                        }
                                    })
                            .flatMap(new UserlabelPartExpandRichFlatMapFunc())
                            .filter(
                                    new FilterFunction<String>() {
                                        @Override
                                        public boolean filter(String value) throws Exception {
                                            JSONObject jsonObject = JSONObject.parseObject(value);
                                            if (jsonObject.containsKey(OUTPUT_KEY_BY)) {
                                                return (jsonObject.get(OUTPUT_KEY_BY) != null);
                                            } else {
                                                return false;
                                            }
                                        }
                                    });

            /** 不拆分高表 宽表输出 */

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
//            sinkProp.setProperty("acks", "all");
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> userLabelAllFlinkKafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(userLabelAllFlinkKafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(DimCallRecordUserInfo.class.toString());
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class UserlabelPartExpandRichFlatMapFunc
            extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;
        HashMap<String, String> mapV1 = new HashMap<>();
        HashMap<String, String> mapV1Clean = new HashMap<>();

        @Override
        public void open(Configuration parameters) throws Exception {
            try {
                userLabelVState =
                        getRuntimeContext()
                                .getState(
                                        new ValueStateDescriptor<JSONObject>(
                                                "userLabelVState", JSONObject.class));
                initV1Map();
                initV1CleanMap();
            } catch (Exception e) {
                LOG.error("UserlabelPartExpandRichFlatMapFunc open exception: " + e.toString());
                e.printStackTrace();
            }
        }

        @Override
        public void flatMap(String input, Collector<String> out) throws Exception {
            try {
                LOG.debug("UserlabelPartExpandRichFlatMapFunc flatMap input: " + input);
                JSONObject tmp = userLabelVState.value();
                JSONObject curJsonObj = (tmp == null ? initJsonObjState(ADMUSERLABELPROP) : tmp);
                JSONObject inputJson = JSON.parseObject(input);
                inputJson = TransformProvCity(inputJson);
                Set<String> keySet = inputJson.keySet();
                for (String key : keySet) {
                    Object value = inputJson.get(key);
                    if (!key.equals(PIDAREA)
                            && !key.equals(ADCOUNTY)
                            && !key.equals(INPUT_KEY_BY)
                            && curJsonObj.containsKey(key)) {
                        curJsonObj.put(key, value);
                    }
                }
                userLabelVState.update(curJsonObj);
                LOG.debug(
                        "UserlabelPartExpandRichFlatMapFunc outJson: "
                                + curJsonObj.toString(SerializerFeature.WriteMapNullValue));
                out.collect(curJsonObj.toString());
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("UserlabelPartExpandRichFlatMapFunc exception input: " + input);
                e.printStackTrace();
            }
        }

        private void initV1Map() throws IOException {
            File v1File =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.V1CATCHFILENAME);
            List<String> v1Lines = FileUtils.readLines(v1File);
            for (String line : v1Lines) {
                String[] splits = line.split(",");
                if (splits.length == 6) {
                    String raw_prov = splits[0];
                    String raw_city = splits[1];
                    String raw_area = splits[2];
                    String new_prov = splits[3];
                    String new_city = splits[4];
                    String new_area = splits[5];
                    String mapKey = raw_prov + SPLITWORD + raw_city + SPLITWORD + raw_area;
                    String mapVal = new_prov + SPLITWORD + new_city + SPLITWORD + new_area;
                    mapV1.put(mapKey, mapVal);
                } else {
                    LOG.error("UserlabelPartExpandRichFlatMapFunc open error line: " + line);
                }
            }
        }

        private void initV1CleanMap() throws IOException {
            File v1CleanFile =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.V1CLEANCATCHFILENAME);
            List<String> v1CleanLines = FileUtils.readLines(v1CleanFile);
            for (String line : v1CleanLines) {
                String[] splits = line.split(",");
                if (splits.length == 4) {
                    String new_prov = splits[0];
                    String new_city = splits[1];
                    String raw_prov = splits[2];
                    String raw_city = splits[3];
                    String mapKey = raw_prov + SPLITWORD + raw_city;
                    String mapVal = new_prov + SPLITWORD + new_city;
                    mapV1Clean.put(mapKey, mapVal);
                } else {
                    LOG.error("UserlabelPartExpandRichFlatMapFunc open error line: " + line);
                }
            }
        }

        private JSONObject TransformProvCity(JSONObject inputJson) {
            try {
                if (inputJson != null) {
                    if (inputJson.containsKey(PIDPROV)) {
                        String pid_prov = (String) inputJson.getOrDefault(PIDPROV, "");
                        String pid_city = (String) inputJson.getOrDefault(PIDCITY, "");
                        String pid_area = (String) inputJson.getOrDefault(PIDAREA, "");

                        String mapkey = pid_prov + SPLITWORD + pid_city + SPLITWORD + pid_area;
                        String mapVal = mapV1.get(mapkey);
                        if (mapVal != null) {
                            String[] splits = mapVal.split(SPLITWORD);
                            if (splits.length == 3) {
                                inputJson.put(PIDPROV, splits[0]);
                                inputJson.put(PIDCITY, splits[1]);
                                inputJson.put(PIDAREA, splits[2]);
                            } else {
                                LOG.error("UserExpandRichFlMp PID Transf err Val: " + mapVal);
                            }
                        } else {
                            LOG.warn("UserExpandRichFlMp PID Transf err Key: " + mapkey);
                            inputJson.put(PIDPROV, "null");
                            inputJson.put(PIDCITY, "null");
                            inputJson.put(PIDAREA, "null");
                        }
                        /** 没关联就舍弃 */
                    } else if (inputJson.containsKey(ADPROV)) {
                        String ad_prov = (String) inputJson.getOrDefault(ADPROV, "");
                        String ad_city = (String) inputJson.getOrDefault(ADCITY, "");
                        String ad_county = (String) inputJson.getOrDefault(ADCOUNTY, "");

                        String mapkey = ad_prov + SPLITWORD + ad_city + SPLITWORD + ad_county;
                        String mapVal = mapV1.get(mapkey);
                        if (mapVal != null) {
                            String[] splits = mapVal.split(SPLITWORD);
                            if (splits.length == 3) {
                                inputJson.put(ADPROV, splits[0]);
                                inputJson.put(ADCITY, splits[1]);
                                inputJson.put(ADCOUNTY, splits[2]);
                            } else {
                                LOG.error("UserExpandRichFlMp AD Transf err Val: " + mapVal);
                            }
                        } else {
                            LOG.warn("UserExpandRichFlMp AD Transf err Key: " + mapkey);
                            inputJson.put(ADPROV, "null");
                            inputJson.put(ADCITY, "null");
                            inputJson.put(ADCOUNTY, "null");
                        }

                    } else if (inputJson.containsKey(MOBPROV)) {
                        String mob_prov = (String) inputJson.getOrDefault(MOBPROV, "");
                        String mob_city = (String) inputJson.getOrDefault(MOBCITY, "");
                        String mob_isp = (String) inputJson.getOrDefault(MOBISP, "");

                        String mapkey = mob_prov + SPLITWORD + mob_city;
                        String mapVal = mapV1Clean.get(mapkey);
                        if (mapVal != null) {
                            String[] splits = mapVal.split(SPLITWORD);
                            if (splits.length == 2) {
                                inputJson.put(MOBPROV, splits[0]);
                                inputJson.put(MOBCITY, splits[1]);
                                inputJson.put(MOBISP, mob_isp);
                            } else {
                                LOG.error("UserExpandRichFlMp MOB Transf err Val: " + mapVal);
                            }
                        } else {
                            LOG.warn("UserExpandRichFlMp MOB Transf err Key: " + mapkey);
                            inputJson.put(MOBPROV, "null");
                            inputJson.put(MOBCITY, "null");
                        }
                    }
                    return inputJson;
                } else {
                    LOG.error("UserExpandRichFlMp Transf input is null");
                }
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("UserExpandRichFlMp Transf err input: " + inputJson);
                e.printStackTrace();
            }
            return inputJson;
        }
    }

    private static class UserLabelRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userLabelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "userLabelVState", JSONObject.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                LOG.debug("Debug UserLabelRichFlatMapFunc flatmap input: " + value);
                JSONObject tmp = userLabelVState.value();
                JSONObject curState = (tmp == null ? initJsonObjState(ADMUSERLABELPROP) : tmp);
                JSONObject input = JSON.parseObject(value);
                String userKeyVal = input.getString(OUTPUT_KEY_BY);
                for (Map.Entry<String, Object> inputEle : input.entrySet()) {
                    String key = inputEle.getKey();
                    Object val = inputEle.getValue();

                    if (!key.equals(OUTPUT_KEY_BY)) {
                        Object jsonVal = curState.get(key);
                        if (jsonVal != null && val != null) {
                            if (!jsonVal.equals(val)) {
                                curState.put(key, val);
                                JSONObject jsonObject = new JSONObject();
                                jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                                jsonObject.put(key, val);
                                out.collect(jsonObject.toString());
                            }
                        } else {
                            if (jsonVal != val) {
                                curState.put(key, val);
                                JSONObject jsonObject = new JSONObject();
                                jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                                jsonObject.put(key, val);
                                out.collect(jsonObject.toString());
                            }
                        }
                    }
                }
                userLabelVState.update(curState);
            } catch (Exception e) {
                LOG.error(e.toString());
                LOG.error("UserLabelRichFlatMapFunc exception input: " + value);
                e.printStackTrace();
            }
        }
    }

    private static JSONObject initJsonObjState(String propFile) throws IOException {

        JSONObject res = new JSONObject();
        InputStream resourceAsStream =
                DimCallRecordUserInfo.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        Set<Object> propSets = reader.getProperties().keySet();
        for (Object key : propSets) {
            String typeValue = reader.get((String) key);
            String[] splits = typeValue.split(":");
            if (splits.length == 2) {
                String type = splits[0];
                String value = splits[1];
                jsonObjectSetValue(res, type, (String) key, value);
            } else {
                LOG.error("split Error: " + splits.length);
            }
        }
        return res;
    }

    private static void jsonObjectSetValue(
            JSONObject jsonObj, String type, String key, String value) {
        if (!value.equals("null")) {
            switch (type) {
                case "Long":
                    jsonObj.put(key, Long.parseLong(value));
                    break;
                case "String":
                    jsonObj.put(key, value);
                    break;
                case "Float":
                    jsonObj.put(key, Float.parseFloat(value));
                    break;
                case "Integer":
                    jsonObj.put(key, Integer.parseInt(value));
                    break;
                default:
                    LOG.error("Not support this type: " + type);
                    break;
            }
        } else {
            jsonObj.put(key, null);
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String key = jsonObject.getString(OUTPUT_KEY_BY);
            return key.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
